package com.we.risk.sa.offline;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
import org.apache.flink.streaming.util.serialization.KeyedSerializationSchema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableConfig;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.catalog.hive.HiveCatalog;
import org.apache.flink.util.Collector;

import com.alibaba.fastjson.JSONObject;
import com.we.flink.utils.WeKafkaPropertyReader;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.InputStream;
import java.sql.Timestamp;
import java.text.SimpleDateFormat;
import java.util.*;

public class SaEventLabelOffline {
    public static final String RELEASEPROP = "risk/saevent_kfk_prod_offline_be.properties";
    public static final String PROPSETS = "risk/saevent/stagefirst/saevent_label.properties";
    public static final String KEYBY = "user_key";
    public static final String SEPARATOR = "#";
    public static final long MILLSECONDSOFDAY = 24 * 3600 * 1000;
    public static final long TIMEZONEDIFF = 8 * 3600 * 1000;
    public static Logger LOG = LoggerFactory.getLogger(SaEventLabelOffline.class);

    public static final String EVTIDFV = "IdentificationView";
    public static final String EVTCPTC = "ChoiceProtocol";
    public static final String EVTITMC = "ItemChoice";
    public static final String EVTITMINP = "ItemInput";
    public static final String EVTBKBT = "BackButton";

    public static final String EVTAPSTT = "$AppStart";
    public static final String EVTAPVSC = "$AppViewScreen";
    public static final String EVTAPCLI = "$AppClick";
    public static final String EVTAPEND = "$AppEnd";
    /** PageTitile */
    public static final String PGTTBDCAD = "绑定银行卡页";

    public static final String PGTTADCAD = "添加银行卡页";
    public static final String PGTTFCIDT = "人脸识别页";
    public static final String PGTTIDPH = "身份证拍照页";
    public static final String PGTTTCID = "腾讯身份认证页";
    public static final String PGTTIDPG = "身份证页";
    public static final String PGTTEMCT = "紧急联系人页";

    /** LocationPage */
    public static final String LCPGBDCAD = "绑定银行卡";

    public static final String LCPGADCAD = "添加银行卡";
    public static final String LCPGFCIDT = "人脸识别页";
    public static final String LCPGBSIF = "基本信息页";
    public static final String LCPGEMCT = "紧急联系人页";

    /** Item */
    public static final String ITEMIDST = "从事行业";

    public static final String ITEMJOB = "工作岗位";
    public static final String ITEMMRG = "婚姻状况";
    public static final String ITEMUSG = "借款用途";
    public static final String ITEMEDU = "最高学历";
    public static final String ITEMICOM = "月收入";

    public static final String DATEFORMAT = "yyyy-MM-dd";

    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        TableConfig tableConf = tableEnv.getConfig();
        Configuration selfConf = new Configuration();
        selfConf.setBoolean("table.exec.hive.infer-source-parallelism", false);
        tableConf.addConfiguration(selfConf);

        try {
            WeKafkaPropertyReader propReader = WeKafkaPropertyReader.init(RELEASEPROP);
            /** RocksDB */
            env.setStateBackend(new RocksDBStateBackend(propReader.getRocksDBBackendUrl()));
            /** checkpoint configure */
            CheckpointConfig ckConf = env.getCheckpointConfig();
            ckConf.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
            ckConf.setCheckpointInterval(10 * 60 * 1000); // ms
            ckConf.setCheckpointTimeout(60 * 60 * 1000);
            ckConf.setMaxConcurrentCheckpoints(1);
            ckConf.setMinPauseBetweenCheckpoints(500);
            ckConf.enableExternalizedCheckpoints(
                    CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

            ckConf.enableUnalignedCheckpoints();

            /** hive catlog configure */
            String hiveCatlogName = propReader.getHiveCatlogName();
            String hiveDBName = propReader.getHiveDBName();
            String hiveConfDir = propReader.getHiveConfDir();

            HiveCatalog hiveCatalog = new HiveCatalog(hiveCatlogName, hiveDBName, hiveConfDir);
            tableEnv.registerCatalog("hive", hiveCatalog);
            tableEnv.useCatalog("hive");

            Table table = tableEnv.sqlQuery(TableSaEvent.QUERYSQL);
            DataStream<TableSaEvent> input = tableEnv.toDataStream(table, TableSaEvent.class);
            SingleOutputStreamOperator<String> out =
                    input.map(
                            new MapFunction<TableSaEvent, String>() {
                                @Override
                                public String map(TableSaEvent value) throws Exception {
                                    long bev_time = value.getBev_time().getTime();
                                    String user_key = value.getDistinct_id();
                                    String event = value.getEvent();
                                    String properties = value.getProperties();
                                    Timestamp recv_time = value.getRecv_time();
                                    String dt = value.getDt();

                                    String res =
                                            "bev_time: "
                                                    + bev_time
                                                    + " user_key: "
                                                    + user_key
                                                    + " event: "
                                                    + event
                                                    + " recv_time: "
                                                    + recv_time
                                                    + " dt: "
                                                    + dt
                                                    + " properties: "
                                                    + properties;

                                    System.out.println(res);

                                    return res;
                                }
                            });
            //            SingleOutputStreamOperator<String> out = input
            //                    .filter(new SaEventFilterFunc())
            //                    .keyBy(new KeySelector<String, String>() {
            //                        @Override
            //                        public String getKey(String value) throws Exception {
            //                            JSONObject inputJson = JSONObject.parseObject(value);
            //                            String user_key = inputJson.getString("distinct_id");
            //                            return user_key;
            //                        }
            //                    }).flatMap(new SaEventFlatMapFunc());
            //
            //            /**
            //             * sink to Kafka
            //             */
            //            String sinkkfkTopic = propReader.getKfkTopic();
            //            Properties sinkProp = new Properties();
            //            sinkProp.setProperty("bootstrap.servers",
            // propReader.getKfkBootStrapServer());
            //            int sinkkfkPartitions = propReader.getKfkPartitions();
            //
            //            FlinkKafkaProducer<String> kafkaProducer = new
            // FlinkKafkaProducer<String>(sinkkfkTopic,
            //                    new WeKafkaKeyedSerializationSchema(),
            //                    sinkProp,
            //                    Optional.of(new WeKafkaCustomPartitioner()));
            //
            //            out.addSink(kafkaProducer).setParallelism(sinkkfkPartitions);

            env.execute(SaEventLabelOffline.class.toString());

        } catch (Exception e) {
            System.out.println(e.getMessage());
            LOG.error(e.getMessage());
            e.printStackTrace();
        }
    }

    private static class SaEventFilterFunc implements FilterFunction<String> {
        @Override
        public boolean filter(String value) throws Exception {
            boolean res =
                    value.contains(EVTIDFV) || value.contains(EVTCPTC) || value.contains(EVTBKBT);
            res = (res || value.contains(EVTITMC) || value.contains(EVTITMINP));
            res =
                    (res
                            || value.contains(EVTAPCLI)
                            || value.contains(EVTAPSTT)
                            || value.contains(EVTAPEND)
                            || value.contains(EVTAPVSC));

            System.out.println("#filter res:" + res);
            return res;
        }
    }

    private static class SaEventFlatMapFunc extends RichFlatMapFunction<String, String> {
        SimpleDateFormat sdf;
        /** 只存近30天内的数据 */
        MapState<Long, String> resMState;

        HashMap<String, String> educationMap;
        HashMap<String, String> idstMap;
        HashMap<String, String> jobMap;
        HashMap<String, String> mrgMap;
        HashMap<String, String> usageMap;
        HashMap<String, Integer> incomMap;

        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            sdf = new SimpleDateFormat(DATEFORMAT);
            educationMap = initEduMap();
            idstMap = initindustryMap();
            jobMap = initJobMap();
            mrgMap = initMrgMap();
            usageMap = initUsageMap();
            incomMap = initIncomeMap();

            resMState =
                    getRuntimeContext()
                            .getMapState(
                                    new MapStateDescriptor<Long, String>(
                                            "resMState", Long.class, String.class));
        }

        @Override
        public void flatMap(String value, Collector<String> out) throws Exception {
            long now = System.currentTimeMillis();
            JSONObject inputJson = JSONObject.parseObject(value);

            Long recvTime = inputJson.getLong("recv_time");
            String userKey = inputJson.getString("distinct_id");
            JSONObject properties = inputJson.getJSONObject("properties");
            String app_version = properties.getString("$app_version");

            System.out.println(
                    "input recvTime: "
                            + (recvTime == null ? "null" : recvTime)
                            + " userkey:"
                            + userKey
                            + " appVersion: "
                            + app_version
                            + " event:"
                            + inputJson.getString("event")
                            + " bevtime: "
                            + inputJson.getLongValue("time"));
            System.out.println(
                    "input PageTitle: "
                            + properties.getString("PageTitle")
                            + " LocationPage: "
                            + properties.getString("LocationPage"));
            /** 脏数据过滤 */
            if (recvTime != null
                    && !recvTime.equals("")
                    && (app_version != null
                            && (app_version.equals("NULL") || app_version.length() == 5))) {

                String event = inputJson.getString("event");
                Long bevTime = inputJson.getLong("time");
                //                String bevDate = sdf.format(new Date(bevTime));
                String pageTitle = properties.getString("PageTitle");
                String locationPage = properties.getString("LocationPage");

                long mapKey = getStartTimeOfDay(bevTime);
                long earliestTime = timeBeforeDays(recvTime, 30);
                System.out.println(
                        " earliestTime: " + earliestTime + " res: " + (bevTime >= earliestTime));
                if (bevTime >= earliestTime) {
                    /** keep 30days data in state */
                    cleanState(resMState, earliestTime);

                    String res = resMState.get(mapKey);
                    JSONObject resJson = null;
                    if (res != null) {
                        resJson = JSONObject.parseObject(res);
                    } else {
                        resJson = initJsonObjState(PROPSETS);
                        resMState.put(mapKey, resJson.toString());
                    }

                    if (event.equals(EVTIDFV)) {
                        switch (pageTitle) {
                            case PGTTIDPH: // step1
                            case PGTTTCID:
                            case PGTTIDPG:
                                setStepPageRange(bevTime, resJson, "step1");
                                break;
                            case PGTTFCIDT: // step3
                                setStepPageRange(bevTime, resJson, "step3");
                                break;
                            case PGTTADCAD: // step5
                            case PGTTBDCAD:
                                setStepPageRange(bevTime, resJson, "step5");
                                break;
                            case PGTTEMCT: // step9
                                setStepPageRange(bevTime, resJson, "step9");
                                break;
                            default:
                                LOG.warn(
                                        "Event:"
                                                + EVTIDFV
                                                + " not support pageTitle: "
                                                + pageTitle
                                                + " now!!");
                                break;
                        }
                    } else if (event.equals(EVTBKBT)) {
                        if (locationPage.equals(LCPGFCIDT)) {
                            // step3
                            setStepPageRange(bevTime, resJson, "step3");
                        }
                    } else if (event.equals(EVTCPTC)) {
                        // step5
                        if (locationPage.equals(LCPGADCAD) || locationPage.equals(LCPGBDCAD)) {
                            setStepPageRange(bevTime, resJson, "step5");
                        }
                    } else if (event.equals(EVTITMINP)) {
                        // step9
                        if (locationPage.equals(LCPGEMCT)) {
                            setStepPageRange(bevTime, resJson, "step9");
                        }
                    } else if (event.equals(EVTITMC)) {
                        if (locationPage.equals(LCPGBSIF)) {
                            String item = properties.getString("item");
                            String operationContent = properties.getString("OperationContent");
                            if (item != null
                                    && (operationContent != null && !operationContent.equals(""))) {
                                switch (item) {
                                    case ITEMEDU: // 学历
                                        String eduCode = educationMap.get(operationContent);
                                        if (eduCode != null) {
                                            resJson.put(eduCode, 1);
                                        }
                                        break;
                                    case ITEMICOM: // 收入均值 + 最后一次填写的收入值 不区分时间戳，
                                        Integer income = incomMap.get(operationContent);
                                        if (income != null) {
                                            int ma_varnum_income_sum =
                                                    resJson.getIntValue("ma_varnum_income_sum");
                                            int ma_varnum_income_cnt =
                                                    resJson.getIntValue("ma_varnum_income_cnt");
                                            ma_varnum_income_sum = ma_varnum_income_sum + income;
                                            ma_varnum_income_cnt += 1;
                                            resJson.put(
                                                    "ma_varnum_income_sum", ma_varnum_income_sum);
                                            resJson.put(
                                                    "ma_varnum_income_cnt", ma_varnum_income_cnt);

                                            long ma_varnum_income_final_time =
                                                    resJson.getLongValue(
                                                            "ma_varnum_income_final_time");
                                            if (bevTime > ma_varnum_income_final_time) {
                                                resJson.put("ma_varnum_income_final_time", bevTime);
                                                resJson.put("ma_varnum_income_final", income);
                                            }
                                        }
                                        break;
                                    case ITEMIDST: // 行业
                                        String idstCode = idstMap.get(operationContent);
                                        if (idstCode != null) {
                                            resJson.put(idstCode, 1);
                                        }
                                        break;
                                    case ITEMJOB: // 工作
                                        String jobCode = jobMap.get(operationContent);
                                        if (jobCode != null) {
                                            resJson.put(jobCode, 1);
                                        }
                                        break;
                                    case ITEMMRG: // 婚姻
                                        String mrgCode = mrgMap.get(operationContent);
                                        if (mrgCode != null) {
                                            resJson.put(mrgCode, 1);
                                        }
                                        break;
                                    case ITEMUSG: // 用途
                                        String usageCode = usageMap.get(operationContent);
                                        if (usageCode != null) {
                                            resJson.put(usageCode, 1);
                                        }
                                        break;
                                    default:
                                        System.out.println(
                                                "Item Error: "
                                                        + event
                                                        + " "
                                                        + locationPage
                                                        + " "
                                                        + item
                                                        + " not support now!!");
                                        LOG.warn(
                                                "Item Error: "
                                                        + event
                                                        + " "
                                                        + locationPage
                                                        + " "
                                                        + item
                                                        + " not support now!!");
                                        break;
                                }
                            }
                        }
                    } else if (event.equals(EVTAPSTT)
                            || event.equals(EVTAPVSC)
                            || event.equals(EVTAPCLI)
                            || event.equals(EVTAPEND)) {
                        // Last OS
                        String os = properties.getString("$os");
                        long ma_varnum_latest_os_time =
                                resJson.getLongValue("ma_varnum_latest_os_time");
                        if (ma_varnum_latest_os_time < bevTime) {
                            resJson.put("ma_varnum_latest_os_time", bevTime);
                            resJson.put("ma_varnum_latest_os", os);
                        }

                    } else {
                        System.out.println("Error Event: " + event + " not support now!!");
                        LOG.error("Error Event: " + event + " not support now!!");
                    }

                    resJson.put("user_key", userKey);
                    long max_bev_time = resJson.getLongValue("max_bev_time");
                    long max_receive_time = resJson.getLongValue("max_receive_time");
                    if (max_bev_time < bevTime) {
                        resJson.put("max_bev_time", bevTime);
                    }
                    if (max_receive_time < recvTime) {
                        resJson.put("max_receive_time", recvTime);
                    }
                    long label_output_time = System.currentTimeMillis();
                    resJson.put("label_output_time", label_output_time);
                    //                    System.out.println("label cost time: " +
                    // (label_output_time - now) + " ms" + " output-recvtime: " + (label_output_time
                    // - recvTime) + " ms" );

                    String resStr = resJson.toString();
                    out.collect(resStr);
                    //
                    // out.collect(resJson.toString(SerializerFeature.WriteMapNullValue));
                    resMState.put(mapKey, resStr);
                }

            } else {
                System.out.println("Warn: filter data!!");
            }
        }

        private HashMap<String, String> initEduMap() {
            HashMap<String, String> resMap = new HashMap<>();
            resMap.put("大学专科和专科学校", "ma_varnum_education_code01");
            resMap.put("初中", "ma_varnum_education_code02");
            resMap.put("技术学校", "ma_varnum_education_code03");
            resMap.put("中等专业学校或中等技术学校", "ma_varnum_education_code04");
            resMap.put("高中", "ma_varnum_education_code05");
            resMap.put("研究生", "ma_varnum_education_code06");
            resMap.put("大学本科", "ma_varnum_education_code07");
            resMap.put("小学", "ma_varnum_education_code08");
            return resMap;
        }

        private HashMap<String, Integer> initIncomeMap() {
            HashMap<String, Integer> resMap = new HashMap<>();
            resMap.put("小于1000元", 500);
            resMap.put("1000-2000元", 1500);
            resMap.put("2000-4000元", 3000);
            resMap.put("4000-6000元", 5000);
            resMap.put("6000-10000元", 8000);
            resMap.put("10000元以上", 10000);
            return resMap;
        }

        private HashMap<String, String> initJobMap() {
            HashMap<String, String> resMap = new HashMap<>();
            resMap.put("自由职业", "ma_varnum_job_code01");
            resMap.put("企业主", "ma_varnum_job_code02");
            resMap.put("工薪族", "ma_varnum_job_code03");
            resMap.put("事业单位公务员", "ma_varnum_job_code04");
            resMap.put("个体户", "ma_varnum_job_code05");
            return resMap;
        }

        private HashMap<String, String> initMrgMap() {

            HashMap<String, String> resMap = new HashMap<>();
            resMap.put("未婚", "ma_varnum_marridge_code01");
            resMap.put("已婚已育", "ma_varnum_marridge_code02");
            resMap.put("已婚未育", "ma_varnum_marridge_code03");
            resMap.put("离异", "ma_varnum_marridge_code04");
            resMap.put("丧偶", "ma_varnum_marridge_code05");
            return resMap;
        }

        private HashMap<String, String> initUsageMap() {
            HashMap<String, String> resMap = new HashMap<>();
            resMap.put("装修", "ma_varnum_usage_code01");
            resMap.put("其他", "ma_varnum_usage_code02");
            resMap.put("教育", "ma_varnum_usage_code03");
            resMap.put("租房", "ma_varnum_usage_code04");
            resMap.put("健康医疗", "ma_varnum_usage_code05");
            resMap.put("手机数码", "ma_varnum_usage_code06");
            resMap.put("家用电器", "ma_varnum_usage_code07");
            resMap.put("旅游", "ma_varnum_usage_code08");
            resMap.put("家具家居", "ma_varnum_usage_code09");
            resMap.put("婚庆", "ma_varnum_usage_code10");
            return resMap;
        }

        private HashMap<String, String> initindustryMap() {
            HashMap<String, String> resMap = new HashMap<>();
            resMap.put("服务业", "ma_varnum_has_industry_code01");
            resMap.put("制造业", "ma_varnum_has_industry_code02");
            resMap.put("建筑工程业", "ma_varnum_has_industry_code03");
            resMap.put("商业", "ma_varnum_has_industry_code04");
            resMap.put("农、林、牧、渔、水利业及其他", "ma_varnum_has_industry_code05");
            resMap.put("通信、电子、互联网行业", "ma_varnum_has_industry_code06");
            resMap.put("交通运输业", "ma_varnum_has_industry_code07");
            resMap.put("餐旅业", "ma_varnum_has_industry_code08");
            resMap.put("政府、非盈利机构", "ma_varnum_has_industry_code09");
            resMap.put("传媒、娱乐、体育、咨询", "ma_varnum_has_industry_code10");
            resMap.put("金融业", "ma_varnum_has_industry_code11");
            resMap.put("文教", "ma_varnum_has_industry_code12");
            resMap.put("公共事业", "ma_varnum_has_industry_code13");
            resMap.put("卫生", "ma_varnum_has_industry_code14");
            resMap.put("家庭管理", "ma_varnum_has_industry_code15");
            resMap.put("新闻、出版、广告业", "ma_varnum_has_industry_code16");
            resMap.put("治安人员", "ma_varnum_has_industry_code17");
            resMap.put("军人", "ma_varnum_has_industry_code18");

            return resMap;
        }

        private void setStepPageRange(Long bevTime, JSONObject resJson, String step) {
            String minKey = "ma_varnum_" + step + "_min_time";
            String maxKey = "ma_varnum_" + step + "_max_time";

            long minTime = resJson.getLongValue(minKey);
            long maxTime = resJson.getLongValue(maxKey);

            if (minTime == 0 || bevTime < minTime) {
                resJson.put(minKey, bevTime);
            }
            if (bevTime > maxTime) {
                resJson.put(maxKey, bevTime);
            }
        }

        //        /**
        //         * use getStartTimeOfDay(bevtime) as state key
        //         * @param bevTime
        //         * @param recvTime
        //         * @param state
        //         * @throws Exception
        //         */
        //        private void updateMstate(long bevTime, long recvTime, MapState<Long, String>
        // state) throws Exception {
        //            long key = getStartTimeOfDay(bevTime);
        //            long earliestTime = timeBeforeDays(recvTime, 30);
        //
        //            if(bevTime >= earliestTime) {
        //                /**
        //                 * keep 30days data in state
        //                 */
        //                cleanState(state, earliestTime);
        //
        //                String range = state.get(key);
        //                if (range != null && range.contains(SEPARATOR)) {
        //                    String[] splits = range.split(SEPARATOR);
        //                    Long min = Long.valueOf(splits[0]);
        //                    Long max = Long.valueOf(splits[1]);
        //                    min = (min < bevTime) ? min : bevTime;
        //                    max = (max > bevTime) ? max : bevTime;
        //                    state.put(key, min + SEPARATOR + max);
        //                } else {
        //                    state.put(key, bevTime + SEPARATOR + bevTime);
        //                }
        //            }
        //        }

        private void cleanState(MapState<Long, String> state, long earliestTime) throws Exception {
            if (!state.isEmpty()) {
                ArrayList<Long> timeLists = new ArrayList<>();
                Iterator<Long> iterator = state.keys().iterator();
                while (iterator.hasNext()) {
                    Long timeKey = iterator.next();
                    if (timeKey < earliestTime) {
                        timeLists.add(timeKey);
                    }
                }
                int listSize = timeLists.size();
                for (int i = 0; i < listSize; i++) {
                    state.remove(timeLists.get(i));
                }
            }
        }

        private long timeBeforeDays(long now, int days) {
            long result = now - (now % MILLSECONDSOFDAY) - (days * MILLSECONDSOFDAY) - TIMEZONEDIFF;
            return result;
        }

        private long getStartTimeOfDay(long now) {
            long result = now - (now % MILLSECONDSOFDAY) - TIMEZONEDIFF;
            return result;
        }

        private static JSONObject initJsonObjState(String propFile) throws IOException {
            JSONObject res = new JSONObject();
            InputStream resourceAsStream =
                    SaEventLabelOffline.class.getClassLoader().getResourceAsStream(propFile);
            ParameterTool reader = ParameterTool.fromPropertiesFile(resourceAsStream);
            Set<Object> propSets = reader.getProperties().keySet();

            for (Object key : propSets) {
                String typeValue = reader.get((String) key);
                String[] splits = typeValue.split(":");
                if (splits.length == 2) {
                    String type = splits[0];
                    String value = splits[1];
                    jsonObjectSetValue(res, type, (String) key, value);
                } else {
                    LOG.error("split Error: " + splits.length);
                }
            }
            return res;
        }

        private static void jsonObjectSetValue(
                JSONObject jsonObj, String type, String key, String value) {
            if (!value.equals("null")) {
                switch (type) {
                    case "Byte":
                        jsonObj.put(key, Byte.parseByte(value));
                        break;
                    case "Long":
                        jsonObj.put(key, Long.parseLong(value));
                        break;
                    case "String":
                        jsonObj.put(key, value);
                        break;
                    case "Float":
                        jsonObj.put(key, Float.parseFloat(value));
                        break;
                    case "Integer":
                        jsonObj.put(key, Integer.parseInt(value));
                        break;
                    default:
                        LOG.error("Not support this type: " + type);
                        break;
                }
            } else {
                jsonObj.put(key, null);
            }
        }
    }

    private static class WeKafkaKeyedSerializationSchema
            implements KeyedSerializationSchema<String> {
        @Override
        public byte[] serializeKey(String element) {
            JSONObject jsonObject = JSONObject.parseObject(element);
            String keyby = jsonObject.getString(KEYBY);
            return keyby.getBytes();
        }

        @Override
        public byte[] serializeValue(String element) {
            return element.getBytes();
        }

        @Override
        public String getTargetTopic(String element) {
            return null;
        }
    }

    private static class WeKafkaCustomPartitioner extends FlinkKafkaPartitioner<String> {

        @Override
        public int partition(
                String record, byte[] key, byte[] value, String targetTopic, int[] partitions) {
            int partition = Math.abs(new String(key).hashCode() % partitions.length);
            if (LOG.isDebugEnabled()) {
                LOG.info(
                        " partitions: "
                                + partitions.length
                                + " partition: "
                                + partition
                                + " key: "
                                + new String(key));
            }
            return partition;
        }
    }
}
